use bytes::Bytes;
use std::time::Duration;
use std::{borrow::BorrowMut, sync::Arc};
use tokio::sync::mpsc::Receiver;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use webrtc::ice::agent::Agent;
use webrtc::ice::candidate::candidate_base::unmarshal_candidate;
use webrtc::ice::candidate::Candidate;
use webrtc::ice::state::ConnectionState;
use webrtc::util::Conn;

use super::cache::remove_peer_connection;
use crate::data_channels::{handler::on_message, message::Message};

use super::get_ice_config;

//
pub async fn new_ice_connection(remote_peer_id: &String) -> anyhow::Result<Arc<Agent>> {
    //获取配置
    let config = get_ice_config()?;

    let ice_agent = Arc::new(Agent::new(config).await?);

    let (cannel_tx, mut cannel_rx) = tokio::sync::mpsc::channel::<()>(1);

    let done_tx = cannel_tx.clone();

    ice_agent.on_candidate(Box::new(
        move |c: Option<Arc<dyn Candidate + Send + Sync>>| {
            let done_tx_clone = done_tx.clone();
            Box::pin(async move {
                if c.is_none() {
                    let _ = done_tx_clone.send(()).await;
                } else {
                    tracing::debug!("ice candidate:{}", c.unwrap().marshal());
                }
            })
        },
    ));

    let remote_peer_id2 = remote_peer_id.clone();
    ice_agent.on_connection_state_change(Box::new(move |c: ConnectionState| {
        tracing::debug!("ICE Connection State has changed: {c}");
        if c == ConnectionState::Failed {
            //
            //需要关闭连接
            if let Err(err) = remove_peer_connection(&remote_peer_id2) {
                tracing::error!("移除缓存中的连接失败!{}", err);
            }
        } else if c == ConnectionState::Disconnected {
            //需要关闭连接
            if let Err(err) = remove_peer_connection(&remote_peer_id2) {
                tracing::error!("移除缓存中的连接失败!{}", err);
            }
        }
        Box::pin(async move {})
    }));

    ice_agent.gather_candidates()?;

    tracing::debug!("等待ice服务获取信息.....");
    let _ = cannel_rx.recv().await;
    tracing::debug!("gathering is done");

    Ok(ice_agent)
}

pub async fn create_ice_data_channel(
    ice_agent: Arc<Agent>,
    connection: Arc<dyn Conn + Send + Sync>,
    remote_peer_id: &String,
) -> anyhow::Result<UnboundedSender<Message>> {
    // 创建数据通道
    let (receiver_tx, mut receiver_rx) = tokio::sync::mpsc::unbounded_channel::<Message>();

    //读线程
    let peer_id = remote_peer_id.clone();
    let sender = receiver_tx.clone();
    let conn_rx = Arc::clone(&connection);
    tokio::spawn(async move {
        //
        let mut size_buf: [u8; 8] = [0u8; 8];

        let sender_1 = sender.clone();
        loop {
            tokio::select! {
                result = conn_rx.recv(&mut size_buf)=>{
                    //读取大小
                    let length = usize::from_ne_bytes(size_buf);
                    let mut size = length;
                    tracing::debug!("接收的消息长度:{}", size);

                    let mut buf:Vec<u8> = vec![0u8;size];

                    while size > 0{
                        if size <= 8192{
                            let mut temp_buffer = vec![0u8; size];
                            if let Ok(recv_size) = conn_rx.recv(&mut temp_buffer).await{
                                // tracing::debug!("接收的消息长度---:{},{}", recv_size, size - recv_size);
                                //拷贝
                                buf[(length-size)..(length-size+recv_size)].copy_from_slice(&temp_buffer[0..recv_size]);
                                //
                                size = size - recv_size;
                            }else{
                                tracing::error!("接收消息失败");
                                break;
                            }
                        }else{
                            let mut temp_buffer = vec![0u8; size];
                            if let Ok(recv_size) = conn_rx.recv(&mut temp_buffer).await{
                                // tracing::debug!("接收的消息长度---:{},{}", recv_size, size - recv_size);
                                //拷贝
                                buf[(length-size)..(length-size+recv_size)].copy_from_slice(&temp_buffer[0..recv_size]);
                                //
                                size = size - recv_size;
                            }else{
                                tracing::error!("接收消息失败");
                                break;
                            }
                        }
                    }
                    //二进制
                    if let Ok(message) = Message::try_from(buf) {
                        tracing::debug!("Message:{:?}", message.data.len());
                        match on_message(&peer_id, message).await {
                            Ok(response) => {
                                //查看是否需要返回
                                if let Some(res_msg) = response {
                                    sender_1.send(res_msg);
                                }
                            }
                            Err(err) => {
                                tracing::error!("消息处理失败!{}", err);
                            }
                        }
                    } else {
                        tracing::error!("消息反序列化失败!");
                    }
                }
            }
        }
    });

    let conn_tx = Arc::clone(&connection);

    let remote_peer_id2 = remote_peer_id.clone();
    tokio::spawn(async move {
        let conn_tx1 = Arc::clone(&conn_tx);

        loop {
            tokio::select! {
                // _=cannel_rx.recv()=>{
                //     //退出
                //     tracing::error!("收到退出指令!");
                //     break;
                // }
                message = receiver_rx.recv() => {
                    if let Some(message) = message {
                        //
                        if let Ok(content) = TryInto::<Vec<u8>>::try_into(message){
                            let len = content.len();
                            tracing::debug!("发送的长度:{}",len);
                            if let Ok(_) = conn_tx1.send(&len.to_ne_bytes()).await{
                                if len > 8192 {
                                    let mut size:usize = len;
                                    while size > 0{
                                        if size > 8192{
                                            if let Ok(result)  =  conn_tx1.send(&content[(len-size)..(len-size+8192)]).await{
                                                size = size - result;
                                            }
                                        }else{
                                            if let Ok(result)   =  conn_tx1.send(&content[(len-size)..]).await{
                                                size = size - result;
                                            }
                                        }
                                    }
                                }else{
                                    let result  =  conn_tx1.send(&content[0..]).await;
                                    if let Err(err) = result {
                                        tracing::error!("消息发送失败: {err}");
                                        break;
                                    }else{
                                        tracing::debug!("消息发送成功!");
                                    }
                                }
                            }else{
                                tracing::error!("发送长度失败!");
                                break;
                            }
                        }else{
                            tracing::error!("消息序列化失败!");
                            continue;
                        }
                    }
                }
            }
        }
        //退出关闭连接
        if let Err(err) = remove_peer_connection(&remote_peer_id2) {
            tracing::error!("移除缓存中的连接失败!{}", err);
        }
    });

    // let d1 = Arc::clone(&data_channel);
    // data_channel.on_error(Box::new(move |err| {
    //     tracing::error!("数据通道发生异常!{}", err);
    //     match d1.ready_state() {
    //         RTCDataChannelState::Closing => {
    //             tracing::debug!("数据通道正在关闭!");
    //         }
    //         RTCDataChannelState::Closed => {
    //             tracing::debug!("数据通道已经关闭!");
    //         }
    //         _ => {}
    //     }
    //     Box::pin(async {})
    // }));

    Ok(receiver_tx)
}

pub async fn set_remote_candidate_by_string(
    peer_connection: Arc<Agent>,
    remote_candidates: &Vec<String>,
) -> anyhow::Result<()> {
    //设置远程
    for candidate in remote_candidates {
        if let Ok(candidate) = unmarshal_candidate(candidate.as_str()) {
            let c: Arc<dyn Candidate + Send + Sync> = Arc::new(candidate);
            peer_connection.add_remote_candidate(&c);
        } else {
            tracing::error!("unmarshal_candidate error!");
            break;
        }
    }
    Ok(())
}

pub async fn get_local_description_string(
    ice_agent: Arc<Agent>,
) -> anyhow::Result<serde_json::Value> {
    let (local_ufrag, local_pwd) = ice_agent.get_local_user_credentials().await;

    let candidates: Vec<String> = ice_agent
        .get_local_candidates()
        .await?
        .iter()
        .map(|item| item.marshal())
        .collect();

    let data = serde_json::json!({
        "ufrag": &local_ufrag,
        "pwd": &local_pwd,
        "candidates": candidates
    });
    Ok(data)
}
